package com.study.rabbitmq;

import com.rabbitmq.client.*;
import org.junit.Before;
import org.junit.Test;

import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;

public class MessageReceiver {
    //用于连接 RabbitMQ Server 的实例
    private Connection conn;
    @Before
    public void init() throws IOException, TimeoutException {
        //1 创建一个连接工厂
        ConnectionFactory connectionFactory = new ConnectionFactory();
        //2 完成工厂的基本信息注册
        connectionFactory.setHost("localhost");//127.0.0.1
        connectionFactory.setPort(5672);//rabbitmq的端口
        connectionFactory.setUsername("yunchmall");//用户名
        connectionFactory.setPassword("yunchmall");//密码
        connectionFactory.setVirtualHost("/mqtest");
        //3 创建一个连接
        conn = connectionFactory.newConnection();
    }
    @Test
    public void t0() {
        System.out.println("输出连接信息：" + conn);
    }

    //1 普通模式，消费消息
    @Test
    public void t1() throws IOException, InterruptedException {
        //创建频道
        Channel channel = conn.createChannel();
        String queueName = "simple_queue";//队列名称
        /**
         * 声明从哪个队列接收消息
         * 参数1：队列名称，参数2：是否为持久化队列，参数3：是否独占当前连接，参数4：是否在使用完毕后自动删除，参数5：其他额外参数
         */
        channel.queueDeclare(queueName, true, false, false, null);
        //创建一个消费者实例，重写处理消息的方法
        DefaultConsumer consumer = new DefaultConsumer(channel) {
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                System.out.println("1号消费者接收到消息：" + new String(body, StandardCharsets.UTF_8));
            }
        };
        /**
         * //监听队列，保证第一时间得到消息
         * 参数1：队列名称，参数2：是否自动确认，true表示消息接收到自动项MQ回复确认信息，Rabbitmq将会从队列中删除这条消息
         * 参数3：处理这条消息的消费者实例
         */
        channel.basicConsume(queueName, true, consumer);
        System.out.println("1号消费者已就绪...");
        TimeUnit.MINUTES.sleep(5);//睡眠1分钟
    }
    //2 工作队列模式，运行多个消费者
    @Test
    public void t2() throws IOException, InterruptedException {
        //创建频道
        Channel channel = conn.createChannel();
        String queueName = "simple_queue";//队列名称
        /**
         * 声明从哪个队列接收消息
         * 参数1：队列名称，参数2：是否为持久化队列，参数3：是否独占当前连接，参数4：是否在使用完毕后自动删除，参数5：其他额外参数
         */
        channel.queueDeclare(queueName, true, false, false, null);
        //创建一个消费者实例，重写处理消息的方法
        DefaultConsumer consumer = new DefaultConsumer(channel) {
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                System.out.println("2号消费者接收到消息：" + new String(body, StandardCharsets.UTF_8));
            }
        };
        /**
         * //监听队列，保证第一时间得到消息
         * 参数1：队列名称，参数2：是否自动确认，true表示消息接收到自动项MQ回复确认信息，Rabbitmq将会从队列中删除这条消息
         * 参数3：处理这条消息的消费者实例
         */
        channel.basicConsume(queueName, true, consumer);
        System.out.println("2号消费者已就绪...");
        TimeUnit.MINUTES.sleep(5);//睡眠1分钟
    }
    @Test
    public void t3() throws IOException, InterruptedException {
        //创建频道
        Channel channel = conn.createChannel();
        String queueName = "simple_queue";//队列名称
        /**
         * 声明从哪个队列接收消息
         * 参数1：队列名称，参数2：是否为持久化队列，参数3：是否独占当前连接，参数4：是否在使用完毕后自动删除，参数5：其他额外参数
         */
        channel.queueDeclare(queueName, true, false, false, null);
        //创建一个消费者实例，重写处理消息的方法
        DefaultConsumer consumer = new DefaultConsumer(channel) {
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                System.out.println("3号消费者接收到消息：" + new String(body, StandardCharsets.UTF_8));
            }
        };
        /**
         * //监听队列，保证第一时间得到消息
         * 参数1：队列名称，参数2：是否自动确认，true表示消息接收到自动项MQ回复确认信息，Rabbitmq将会从队列中删除这条消息
         * 参数3：处理这条消息的消费者实例
         */
        channel.basicConsume(queueName, true, consumer);
        System.out.println("3号消费者已就绪...");
        TimeUnit.MINUTES.sleep(5);//睡眠1分钟
    }
    @Test
    public void t4() throws IOException, InterruptedException {
        //创建频道
        Channel channel = conn.createChannel();
        String queueName = "simple_queue";//队列名称
        /**
         * 声明从哪个队列接收消息
         * 参数1：队列名称，参数2：是否为持久化队列，参数3：是否独占当前连接，参数4：是否在使用完毕后自动删除，参数5：其他额外参数
         */
        channel.queueDeclare(queueName, true, false, false, null);
        //创建一个消费者实例，重写处理消息的方法
        DefaultConsumer consumer = new DefaultConsumer(channel) {
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                System.out.println("4号消费者接收到消息：" + new String(body, StandardCharsets.UTF_8));
            }
        };
        /**
         * //监听队列，保证第一时间得到消息
         * 参数1：队列名称，参数2：是否自动确认，true表示消息接收到自动项MQ回复确认信息，Rabbitmq将会从队列中删除这条消息
         * 参数3：处理这条消息的消费者实例
         */
        channel.basicConsume(queueName, true, consumer);
        System.out.println("4号消费者已就绪...");
        TimeUnit.MINUTES.sleep(5);//睡眠1分钟
    }
    //3 接收广播消息
    @Test
    public void t5() throws IOException, InterruptedException {
        //创建频道
        Channel channel = conn.createChannel();
        String queueName = "queue_fanout1";//队列名称，5号消费者接收 queue_fanout1 队列上的消息
        /**
         * 声明从哪个队列接收消息
         * 参数1：队列名称，参数2：是否为持久化队列，参数3：是否独占当前连接，参数4：是否在使用完毕后自动删除，参数5：其他额外参数
         */
        channel.queueDeclare(queueName, true, false, false, null);
        //创建一个消费者实例，重写处理消息的方法
        DefaultConsumer consumer = new DefaultConsumer(channel) {
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                System.out.println("5号消费者接收到消息：" + new String(body, StandardCharsets.UTF_8));
            }
        };
        /**
         * //监听队列，保证第一时间得到消息
         * 参数1：队列名称，参数2：是否自动确认，true表示消息接收到自动项MQ回复确认信息，Rabbitmq将会从队列中删除这条消息
         * 参数3：处理这条消息的消费者实例
         */
        channel.basicConsume(queueName, true, consumer);
        System.out.println("5号消费者已就绪...");
        TimeUnit.MINUTES.sleep(5);//睡眠1分钟
    }
    @Test
    public void t6() throws IOException, InterruptedException {
        //创建频道
        Channel channel = conn.createChannel();
        String queueName = "queue_fanout2";//队列名称，6号消费者接收 queue_fanout2 队列上的消息
        /**
         * 声明从哪个队列接收消息
         * 参数1：队列名称，参数2：是否为持久化队列，参数3：是否独占当前连接，参数4：是否在使用完毕后自动删除，参数5：其他额外参数
         */
        channel.queueDeclare(queueName, true, false, false, null);
        //创建一个消费者实例，重写处理消息的方法
        DefaultConsumer consumer = new DefaultConsumer(channel) {
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                System.out.println("6号消费者接收到消息：" + new String(body, StandardCharsets.UTF_8));
            }
        };
        /**
         * //监听队列，保证第一时间得到消息
         * 参数1：队列名称，参数2：是否自动确认，true表示消息接收到自动项MQ回复确认信息，Rabbitmq将会从队列中删除这条消息
         * 参数3：处理这条消息的消费者实例
         */
        channel.basicConsume(queueName, true, consumer);
        System.out.println("6号消费者已就绪...");
        TimeUnit.MINUTES.sleep(5);//睡眠1分钟
    }
    // 路由模式接收消息
    @Test
    public void t7() throws IOException, InterruptedException {
        //创建频道
        Channel channel = conn.createChannel();
        String queueName = "queue_direct1";//队列名称，7号消费者接收 queue_direct1 队列上的消息
        /**
         * 声明从哪个队列接收消息
         * 参数1：队列名称，参数2：是否为持久化队列，参数3：是否独占当前连接，参数4：是否在使用完毕后自动删除，参数5：其他额外参数
         */
        channel.queueDeclare(queueName, true, false, false, null);
        //创建一个消费者实例，重写处理消息的方法
        DefaultConsumer consumer = new DefaultConsumer(channel) {
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                System.out.println("7号消费者接收到消息：" + new String(body, StandardCharsets.UTF_8));
            }
        };
        /**
         * //监听队列，保证第一时间得到消息
         * 参数1：队列名称，参数2：是否自动确认，true表示消息接收到自动项MQ回复确认信息，Rabbitmq将会从队列中删除这条消息
         * 参数3：处理这条消息的消费者实例
         */
        channel.basicConsume(queueName, true, consumer);
        System.out.println("7号消费者已就绪...");
        TimeUnit.MINUTES.sleep(5);//睡眠1分钟
    }
    @Test
    public void t8() throws IOException, InterruptedException {
        //创建频道
        Channel channel = conn.createChannel();
        String queueName = "queue_direct2";//队列名称，8号消费者接收 queue_direct2 队列上的消息
        /**
         * 声明从哪个队列接收消息
         * 参数1：队列名称，参数2：是否为持久化队列，参数3：是否独占当前连接，参数4：是否在使用完毕后自动删除，参数5：其他额外参数
         */
        channel.queueDeclare(queueName, true, false, false, null);
        //创建一个消费者实例，重写处理消息的方法
        DefaultConsumer consumer = new DefaultConsumer(channel) {
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                System.out.println("8号消费者接收到消息：" + new String(body, StandardCharsets.UTF_8));
            }
        };
        /**
         * //监听队列，保证第一时间得到消息
         * 参数1：队列名称，参数2：是否自动确认，true表示消息接收到自动项MQ回复确认信息，Rabbitmq将会从队列中删除这条消息
         * 参数3：处理这条消息的消费者实例
         */
        channel.basicConsume(queueName, true, consumer);
        System.out.println("8号消费者已就绪...");
        TimeUnit.MINUTES.sleep(5);//睡眠1分钟
    }

    @Test
    public void t9() throws IOException, InterruptedException {
        //创建频道
        Channel channel = conn.createChannel();
        String queueName = "queue_topic1";//队列名称，9号消费者接收 queue_topic1 队列上的消息
//        String queueName = channel.queueDeclare().getQueue();
        //声明交换机
        String exchange = "exchange_topic";
        channel.exchangeDeclare(exchange, BuiltinExchangeType.TOPIC, true, false,false, null);
//        channel.exchangeDeclare(exchange, BuiltinExchangeType.TOPIC);
        /**
         * 声明从哪个队列接收消息
         * 参数1：队列名称，参数2：是否为持久化队列，参数3：是否独占当前连接，参数4：是否在使用完毕后自动删除，参数5：其他额外参数
         */
        channel.queueDeclare(queueName, true, false, false, null);
        //绑定队列，同时设置关注的主题（这样做的目的是可以对当前队列保存的消息进行过滤，可以选择性的消费这个队列中的部分消息）
        channel.queueBind(queueName, exchange, "*.view");

        /*DeliverCallback deliverCallback = (consumerTag, delivery) -> {
            String message = new String(delivery.getBody(), "UTF-8");
            System.out.println(" [x] Received '" + delivery.getEnvelope().getRoutingKey() + "':'" + message + "'");
        };
        channel.basicConsume(queueName, true, deliverCallback, consumerTag -> { });*/
        //创建一个消费者实例，重写处理消息的方法
        DefaultConsumer consumer = new DefaultConsumer(channel) {
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                System.out.println("9号消费者接收到消息：" + new String(body, StandardCharsets.UTF_8));
                System.out.println("消息主题：" + envelope.getRoutingKey());
                System.out.println("交换机信息：" + envelope.getExchange());
            }
        };
        /**
         * //监听队列，保证第一时间得到消息
         * 参数1：队列名称，参数2：是否自动确认，true表示消息接收到自动项MQ回复确认信息，Rabbitmq将会从队列中删除这条消息
         * 参数3：处理这条消息的消费者实例
         */
        channel.basicConsume(queueName, true, consumer);
        System.out.println("9号消费者已就绪...");
        TimeUnit.MINUTES.sleep(5);//睡眠1分钟
    }

}
